package com.tmp.alarm.task;

import com.tmp.alarm.utils.MySQLUtils;
import com.tmp.alarm.utils.PropertiesUtil;
import com.tmp.alarm.utils.UUIDByDateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**计算不同队列中使用的used_memory占分配的memory比例的大小来告警
 */
public class UsedMemAlarm {

    private  final  static Logger LOGGER = LoggerFactory.getLogger(UsedMemAlarm.class);
    private  final  static  String processing_type = PropertiesUtil.get("processing_type").trim();

    /**
     * 报警表queue_memory_alarm中queue_id: 1表示 ge队列,2表示 alliance,3表示 mem,4表示 spark,5表示 dev,6表示 recommed,7表示 analysis
     *
     * 统计不同的队列,并告警,统计查询type=1的值,并且统计完后将type改为type=2
     * ge: 表示root.bdp_jmart_mobile_union.bdp_jmart_mobile_ge队列
     * alliance:表示root.bdp_jmart_mobile_union.bdp_jmart_mobile_alliance队列
     * mem:表示root.bdp_jmart_mobile_union.bdp_jmart_mobile_mem队列
     * spark:表示root.bdp_jmart_mobile_union.bdp_jmart_mobile_spark队列
     * dev:表示root.bdp_jmart_mobile_union.bdp_jmart_mobile_dev队列
     * recommed:表示root.bdp_jmart_mobile_union.bdp_jmart_mobile_recommed队列
     * analysis:表示root.bdp_jmart_mobile_union.bdp_jmart_mobile_analysis队列
     */
    public static  void countmemoryRateAlarm(){

        String currentTime = UUIDByDateUtils.getCurrentTimes("yyyyMMddHHmmss");

        String AllQueueNumSql ="select count(1)  AS sumnums from (select * from mart_mobile_queue_log as m where m.type=1 and m.log_time <="+ currentTime + " group by m.queue_name) as tmp";

        String geSql ="select b.used_memory,b.min_memory from (select queue_name, max(id)as id from mart_mobile_queue_log  where type=1 and log_time <="+ currentTime +" and queue_name like '%ge%') as a,mart_mobile_queue_log as b where a.id =b.id";

        String allianceSql ="select b.used_memory,b.min_memory from (select queue_name, max(id)as id from mart_mobile_queue_log  where type=1 and log_time <="+ currentTime +" and queue_name like '%alliance%') as a,mart_mobile_queue_log as b where a.id =b.id";

        String memSql ="select b.used_memory,b.min_memory from (select queue_name, max(id)as id from mart_mobile_queue_log  where type=1 and log_time <="+ currentTime +" and queue_name like '%mem%') as a,mart_mobile_queue_log as b where a.id =b.id";

        String sparkSql ="select b.used_memory,b.min_memory from (select queue_name, max(id)as id from mart_mobile_queue_log where type=1 and log_time <="+ currentTime +" and queue_name like '%spark%') as a,mart_mobile_queue_log as b where a.id =b.id";

        String devSql ="select b.used_memory,b.min_memory from (select queue_name, max(id)as id from mart_mobile_queue_log where type=1 and log_time <="+ currentTime +" and queue_name like '%dev%') as a,mart_mobile_queue_log as b where a.id =b.id";

        String recommedSql ="select b.used_memory,b.min_memory from (select queue_name, max(id)as id from mart_mobile_queue_log where type=1 and log_time <="+ currentTime +" and queue_name like '%recommed%') as a,mart_mobile_queue_log as b where a.id =b.id";

        String analysisSql ="select b.used_memory,b.min_memory from (select queue_name, max(id)as id from mart_mobile_queue_log where type=1 and log_time <="+ currentTime +" and queue_name like '%analysis%') as a,mart_mobile_queue_log as b where a.id =b.id";

        Connection connection = null;
        String totalQueues="";
        String ge_used_min_memory = "";
        String alliance_used_min_memory="";
        String mem_used_min_memory="";
        String spark_used_min_memory="";
        String dev_used_min_memory="";
        String recommed_used_min_memory="";
        String analysis_used_min_memory="";
        try {
            connection = MySQLUtils.getConnection();
            totalQueues = countQueuesNum(AllQueueNumSql, connection);
            if(Integer.parseInt(totalQueues.trim()) <=0){
                LOGGER.info("本次queue队列总数统计totalQueues查询为0在type=1情况下...");
                return;
            }

            ge_used_min_memory = get_used_min_memory(geSql, connection);
            alliance_used_min_memory = get_used_min_memory(allianceSql, connection);
            mem_used_min_memory = get_used_min_memory(memSql, connection);
            spark_used_min_memory = get_used_min_memory(sparkSql, connection);
            dev_used_min_memory = get_used_min_memory(devSql, connection);
            recommed_used_min_memory = get_used_min_memory(recommedSql, connection);
            analysis_used_min_memory = get_used_min_memory(analysisSql, connection);
        } catch (Exception e) {
            e.printStackTrace();
        }

        //更新已经查询过的记录,将type设置为2,供进度查询使用
        String update_sql = "UPDATE mart_mobile_queue_log SET type="+processing_type+" WHERE type=1";
        MySQLUtils.executeUpdate(connection,update_sql);

        //获取批次号 batch_number
        String batch_number_sql = "SELECT j.id,j.batch_number FROM queue_memory_alarm j WHERE j.queue_id=1 ORDER BY id DESC LIMIT 1";
        Connection conn = null;
        String batch_number = null;
        try {
            conn = MySQLUtils.getConnection();
            ResultSet batch_number_rs = MySQLUtils.query(conn, batch_number_sql);
            batch_number_rs.next();
            int batch_number_rows = batch_number_rs.getRow();
            if (batch_number_rows > 0){
                String batch_number_str = batch_number_rs.getString("batch_number");
                long batch_number_int = Long.parseLong(batch_number_str) + 1;
                batch_number = batch_number_int + "";
                System.out.println("查询的批次号："+batch_number);
            }else {
                batch_number = 1+"";
                System.out.println("首次自定义批次号："+batch_number);
            }
        }catch (Exception e){
            e.printStackTrace();
        }


        //表queue_memory_alarm告警类型;1表示 ge,2表示 alliance,3表示 mem,4表示 spark,5表示 dev,6表示 recommed,7表示 analysis

        //ge 类型告警
        geAlarm(currentTime, ge_used_min_memory,batch_number);

        //alliance 类型告警
        allianceAlarm(currentTime, alliance_used_min_memory,batch_number);

        //mem 类型浏览器界面提交的任务超过一定比例后,告警
        memAlarm(currentTime, mem_used_min_memory,batch_number);

        //spark 类型告警
        sparkAlarm(currentTime, spark_used_min_memory,batch_number);

        //dev 类型告警
        devAlarm(currentTime, dev_used_min_memory,batch_number);

        //recommed 类型告警
        recommedAlarm(currentTime, recommed_used_min_memory,batch_number);

        //analysis 类型告警
        analysisAlarm(currentTime, analysis_used_min_memory,batch_number);


        MySQLUtils.returnConnection(connection);
        MySQLUtils.returnConnection(conn);

//        LOGGER.info(allNums+"  "+buffaloNums+"  "+gateoneNums+"  "+ideonlineNums+"  "+idejobNums);
    }

    /**报警表建表及字段说明
     *
     use test_mobile;
     create table queue_memory_alarm(
     id INT NOT NULL AUTO_INCREMENT,
     queue_id int,
     alarm_time VARCHAR(32),
     alarm_limit_time VARCHAR(32),
     used_memory int,
     total_memory int,
     cal_rate int,
     self_rate int,
     memory_limit int,
     batch_number int,
     is_alarm int,
     PRIMARY KEY ( id )
     );
     *
     */

    /**
     * ge 队列类型统计告警
     * @param currentTime
     * @param ge_used_min_memory
     */
    private static void geAlarm(String currentTime,String ge_used_min_memory,String batch_number) {

        int memory_limit = Integer.parseInt(PropertiesUtil.get("ge_memory_limit").trim());
        int total_memory_limit_ge = Integer.parseInt(PropertiesUtil.get("total_memory_limit_ge"));
        String[] memory = ge_used_min_memory.split("\t");
        String used_mem = memory[0];
        String min_mem = memory[1];
        System.out.println("used_mem="+used_mem+"；min_mem"+min_mem);
        if(Integer.parseInt(used_mem.trim())<memory_limit && Integer.parseInt(used_mem.trim())<Integer.parseInt(min_mem.trim())){
            LOGGER.info("ge队列使用的数没有超过配置中数,current_used_memory:"+used_mem+" config_memory_limit:"+memory_limit+"  min_mem:"+min_mem);
            return;
        }
        float geRate = Float.parseFloat(used_mem)/Float.parseFloat(min_mem);
        String twoPoint_rate = UUIDByDateUtils.twoPointNumbers(geRate+"");
        String rate = twoPoint_rate.replaceAll("\\.", "");
        int cal_rate = Integer.parseInt(rate);
        int self_rate = Integer.parseInt(PropertiesUtil.get("ge_rate").trim());
        String alarm_limit_time = PropertiesUtil.get("queue_alarm_limit_time").trim();
        System.out.println("计算的使用率："+cal_rate+",设置的使用率："+self_rate);//------------------------
        if(cal_rate>self_rate) {
            //告警的手机号
            String alarm_list = PropertiesUtil.get("alarm_phone_list").trim();
            String alarm_title = "ge队列使用memory过高";
            String alarm_content = "ge队列使用memory过高,使用大小:" + used_mem + " ,总数有:" + min_mem + " ,所占比例为:" + rate + "%";

            //判断在告警记录表(job_source_alarm)中是否有同类型的ge告警过,如果alarm_time到当前时间超过了配置中限制时间alarm_limit_time则告警,
            // 否则不同时满足比例大于指定比例,间隔时间不大于指定时间则不告警
            //表jobs_source_alarm告警类型;1表示ge,2表示alliance,3表示mem,4表示spark,5表示dev,6表示recommed,7表示analysis
            String sql = "SELECT j.id,j.alarm_time FROM queue_memory_alarm j WHERE j.is_alarm=1 and j.queue_id=1 ORDER BY id DESC LIMIT 1";
            ResultSet rs = null;
            Connection connection = null;
            try {
                connection = MySQLUtils.getConnection();
                rs = MySQLUtils.query(connection, sql);
                rs.next();
                int rows = rs.getRow();
                //大于0说明jobs_source_alarm中已经告警过该类型的memory,则根据alarm_time与当前时间判断是否超过了配置中设置的间隔时间alarm_limit_time,超过则再一次告警
                //超过后告警成功则把新的告警记录保存下来,没超过就不用管了
                String addSql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(1,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number  + ",1)";
                String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(1,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number  + ",2)";

                if (rows > 0) {
                    String last_alarm_time = rs.getString("alarm_time");
                    System.out.println("上次报警时间："+last_alarm_time);
                    boolean geFlag = jdugeJobTimeOut(currentTime, last_alarm_time, Integer.parseInt(alarm_limit_time.trim()));
                    if (geFlag) {
                        LOGGER.info("ge队列在queue_memory_alarm中有该类型告警记录,本次告警距离上次告警超过了指定间隔时间,将告警:" + alarm_content);
                        try {
                            MySQLUtils.executeUpdate(connection, addSql);
                        }catch (Exception e){
                            LOGGER.error("ge队列告警失败!!!");
                        }
                    }else {//本次告警距离上次告警没超过指定间隔时间，但保存记录alarm_type类型改为2
                        MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
                    }
                }
                else {//该类型的队列还没有告警过,直接告警,并保存告警记录
                    try {
                        LOGGER.info("ge队列该类型首次告警成功...");
                        MySQLUtils.executeUpdate(connection, addSql);
                    }catch (Exception e){
                        LOGGER.error("ge队列告警发送短信失败!!!");
                    }
                }
                MySQLUtils.returnConnection(connection);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(1,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number  + ",2)";
            Connection connection = null;
            try {
                connection = MySQLUtils.getConnection();
                MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
            }catch (Exception e){
                e.printStackTrace();
            }
            LOGGER.info("ge队列本次统计比例没有超过指定的比例...");
        }
    }

    /**
     * alliance 队列类型统计告警
     *
     */

    private static void allianceAlarm(String currentTime,String ge_used_min_memory,String batch_number) {

        int memory_limit = Integer.parseInt(PropertiesUtil.get("alliance_memory_limit").trim());
        int total_memory_limit_alliance = Integer.parseInt(PropertiesUtil.get("total_memory_limit_alliance"));
        String[] memory = ge_used_min_memory.split("\t");
        String used_mem = memory[0];
        String min_mem = memory[1];
        float allianceRate = Float.parseFloat(used_mem)/Float.parseFloat(min_mem);
        String twoPoint_rate = UUIDByDateUtils.twoPointNumbers(allianceRate + "");
        String rate = twoPoint_rate.replaceAll("\\.", "");
        int cal_rate = Integer.parseInt(rate);
        int self_rate = Integer.parseInt(PropertiesUtil.get("alliance_rate").trim());
        String alarm_limit_time = PropertiesUtil.get("queue_alarm_limit_time").trim();
        if(Integer.parseInt(used_mem.trim())<memory_limit && Integer.parseInt(used_mem.trim())<Integer.parseInt(min_mem.trim())){
            LOGGER.info("alliance队列使用的数没有超过配置中数,current_used_memory:"+used_mem+" config_memory_limit:"+ memory_limit +"  min_mem:"+ min_mem);
            String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(2,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number  + ",2)";
            Connection connection = null;
            try {
                connection = MySQLUtils.getConnection();
                MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
            }catch (Exception e){
                e.printStackTrace();
            }
            return;
        }

        if(cal_rate>=self_rate) {
            System.out.println("计算的使用率："+cal_rate+",设置的使用率："+self_rate);//------------------------
            //告警的手机号
            String alarm_list = PropertiesUtil.get("alarm_phone_list").trim();
            String alarm_title = "alliance队列使用memory过高";
            String alarm_content = "alliance队列使用memory过高,使用数:" + used_mem + " ,总数有:" + min_mem + " ,所占比例为:" + rate + "%";

            //判断在告警记录表(job_source_alarm)中是否有同类型的ge告警过,如果alarm_time到当前时间超过了配置中限制时间alarm_limit_time则告警,
            // 否则不同时满足比例大于指定比例,间隔时间不大于指定时间则不告警
            //表jobs_source_alarm告警类型;1表示ge,2表示alliance,3表示mem,4表示spark,5表示dev,6表示recommed,7表示analysis
            String sql = "SELECT j.id,j.alarm_time FROM queue_memory_alarm j WHERE j.is_alarm=1 and j.queue_id=2 ORDER BY id DESC LIMIT 1";
            Connection connection = null;
            ResultSet rs = null;
            try {
                connection = MySQLUtils.getConnection();
                rs = MySQLUtils.query(connection, sql);
                rs.next();
                int rows = rs.getRow();
                //大于0说明jobs_source_alarm中已经告警过该类型的memory,则根据alarm_time与当前时间判断是否超过了配置中设置的间隔时间alarm_limit_time,超过则再一次告警
                //超过后告警成功则把新的告警记录保存下来,没超过就不用管了
                String addSql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(2,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",1)";
                String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(2,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",2)";

                if (rows > 0) {
                    String last_alarm_time = rs.getString("alarm_time");
                    System.out.println("上次报警时间："+last_alarm_time);
                    boolean geFlag = jdugeJobTimeOut(currentTime, last_alarm_time, Integer.parseInt(alarm_limit_time.trim()));
                    if (geFlag) {
                        LOGGER.info("alliance队列在queue_memory_alarm中有该类型告警记录,本次告警距离上次告警超过了指定间隔时间,将告警:" + alarm_content);
                        try {
                            MySQLUtils.executeUpdate(connection, addSql);
                        }catch (Exception e){
                            LOGGER.error("alliance队列告警失败!!!");
                        }
                    }else {//本次告警距离上次告警没超过指定间隔时间，但保存记录alarm_type类型改为2
                        MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
                    }
                }
                else {//该类型的队列还没有告警过,直接告警,并保存告警记录
                    try {
                        LOGGER.info("alliance队列该类型首次告警成功...");
                        MySQLUtils.executeUpdate(connection, addSql);
                    }catch (Exception e){
                        LOGGER.error("alliance队列告警发送短信失败!!!");
                    }
                }
                MySQLUtils.returnConnection(connection);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(2,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number  + ",2)";
            Connection connection = null;
            try {
                connection = MySQLUtils.getConnection();
                MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
            }catch (Exception e){
                e.printStackTrace();
            }
            LOGGER.info("alliance队列本次统计比例没有超过指定的比例...");
        }
    }


    /**
     * core 队列类型统计告警
     * @param currentTime
     * @param ge_used_min_memory
     */
    private static void memAlarm(String currentTime,String ge_used_min_memory,String batch_number) {

        int memory_limit = Integer.parseInt(PropertiesUtil.get("mem_memory_limit").trim());
        int total_memory_limit_mem = Integer.parseInt(PropertiesUtil.get("total_memory_limit_mem"));
        String[] memory = ge_used_min_memory.split("\t");
        String used_mem = memory[0];
        String min_mem = memory[1];
        if(Integer.parseInt(used_mem.trim())<memory_limit && Integer.parseInt(used_mem.trim())<Integer.parseInt(min_mem.trim())){
            LOGGER.info("core队列使用的数没有超过配置中数,current_used_memory:"+used_mem+" config_memory_limit:"+memory_limit+"  min_mem:"+min_mem);
            return;
        }

        float coreRate = Float.parseFloat(used_mem)/Float.parseFloat(min_mem);
        String twoPoint_rate = UUIDByDateUtils.twoPointNumbers(coreRate+"");
        String rate = twoPoint_rate.replaceAll("\\.", "");
        int cal_rate = Integer.parseInt(rate);
        int self_rate = Integer.parseInt(PropertiesUtil.get("core_rate").trim());
        String alarm_limit_time = PropertiesUtil.get("queue_alarm_limit_time").trim();
        if(cal_rate>self_rate) {
            //告警的手机号
            String alarm_list = PropertiesUtil.get("alarm_phone_list").trim();
            String alarm_title = "core队列使用memory过高";
            String alarm_content = "core队列使用memory过高,使用:" + used_mem + " ,总数为:" + min_mem + " ,所占比例为:" + rate + "%";

            //判断在告警记录表(job_source_alarm)中是否有同类型的ge告警过,如果alarm_time到当前时间超过了配置中限制时间alarm_limit_time则告警,
            // 否则不同时满足比例大于指定比例,间隔时间不大于指定时间则不告警
            //表jobs_source_alarm告警类型;1表示ge,2表示alliance,3表示core,4表示spark,5表示dev,6表示recommed,7表示analysis
            String sql = "SELECT j.id,j.alarm_time FROM queue_memory_alarm j WHERE j.is_alarm=1 and j.queue_id=3 ORDER BY id DESC LIMIT 1";
            Connection connection = null;
            ResultSet rs = null;
            try {
                connection = MySQLUtils.getConnection();
                rs = MySQLUtils.query(connection, sql);
                rs.next();
                int rows = rs.getRow();
                //大于0说明jobs_source_alarm中已经告警过该类型的memory,则根据alarm_time与当前时间判断是否超过了配置中设置的间隔时间alarm_limit_time,超过则再一次告警
                //超过后告警成功则把新的告警记录保存下来,没超过就不用管了
                String addSql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(3,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",1)";
                String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(3,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",2)";

                if (rows > 0) {
                    String last_alarm_time = rs.getString("alarm_time");
                    boolean geFlag = jdugeJobTimeOut(currentTime, last_alarm_time, Integer.parseInt(alarm_limit_time.trim()));
                    if (geFlag) {
                        LOGGER.info("core队列在queue_memory_alarm中有该类型告警记录,本次告警距离上次告警超过了指定间隔时间,将告警:" + alarm_content);
                        try {
                            MySQLUtils.executeUpdate(connection, addSql);
                        }catch (Exception e){
                            LOGGER.error("core队列告警失败!!!");
                        }
                    }else {//本次告警距离上次告警没超过指定间隔时间，但保存记录alarm_type类型改为2
                        MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
                    }
                }
                else {//该类型的队列还没有告警过,直接告警,并保存告警记录
                    try {
                        LOGGER.info("core队列该类型首次告警成功...");
                        MySQLUtils.executeUpdate(connection, addSql);
                    }catch (Exception e){
                        LOGGER.error("core队列告警发送短信失败!!!");
                    }
                }
                MySQLUtils.returnConnection(connection);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(3,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",2)";
            Connection connection = null;
            try {
                connection = MySQLUtils.getConnection();
                MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
            }catch (Exception e){
                e.printStackTrace();
            }
            LOGGER.info("core队列本次统计比例没有超过指定的比例...");
        }
    }


    /**
     * spark 队列类型统计告警
     * @param currentTime
     * @param ge_used_min_memory
     */

    private static void sparkAlarm(String currentTime,String ge_used_min_memory,String batch_number) {

        int memory_limit = Integer.parseInt(PropertiesUtil.get("spark_memory_limit").trim());
        int total_memory_limit_spark = Integer.parseInt(PropertiesUtil.get("total_memory_limit_spark"));
        String[] memory = ge_used_min_memory.split("\t");
        String used_mem = memory[0];
        String min_mem = memory[1];
        if(Integer.parseInt(used_mem.trim())<memory_limit && Integer.parseInt(used_mem.trim())<Integer.parseInt(min_mem.trim())){
            LOGGER.info("spark队列使用的数没有超过配置中数,current_used_memory:"+used_mem+" config_memory_limit:"+memory_limit+"  min_mem:"+min_mem);
            return;
        }

        float sparkRate = Float.parseFloat(used_mem)/Float.parseFloat(min_mem);
        String twoPoint_rate = UUIDByDateUtils.twoPointNumbers(sparkRate+"");
        String rate = twoPoint_rate.replaceAll("\\.", "");
        int cal_rate = Integer.parseInt(rate);
        int self_rate = Integer.parseInt(PropertiesUtil.get("spark_rate").trim());
        String alarm_limit_time = PropertiesUtil.get("queue_alarm_limit_time").trim();
        if(cal_rate>self_rate) {
            //告警的手机号
            String alarm_list = PropertiesUtil.get("alarm_phone_list").trim();
            String alarm_title = "spark队列使用memory过高";
            String alarm_content = "spark队列使用memory过高,使用:" + used_mem + " ,总数为:" + min_mem + " ,所占比例为:" + rate + "%";

            //判断在告警记录表(job_source_alarm)中是否有同类型的ge告警过,如果alarm_time到当前时间超过了配置中限制时间alarm_limit_time则告警,
            // 否则不同时满足比例大于指定比例,间隔时间不大于指定时间则不告警
            //表jobs_source_alarm告警类型;1表示ge,2表示alliance,3表示mem,4表示spark,5表示dev,6表示recommed,7表示analysis
            String sql = "SELECT j.id,j.alarm_time FROM queue_memory_alarm j WHERE j.is_alarm=1 and j.queue_id=4 ORDER BY id DESC LIMIT 1";

            Connection connection = null;
            ResultSet rs = null;
            try {
                connection = MySQLUtils.getConnection();
                rs = MySQLUtils.query(connection, sql);
                rs.next();
                int rows = rs.getRow();
                //大于0说明jobs_source_alarm中已经告警过该类型的memory,则根据alarm_time与当前时间判断是否超过了配置中设置的间隔时间alarm_limit_time,超过则再一次告警
                //超过后告警成功则把新的告警记录保存下来,没超过就不用管了
                String addSql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(4,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",1)";
                String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(4,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",2)";

                if (rows > 0) {
                    String last_alarm_time = rs.getString("alarm_time");
                    boolean geFlag = jdugeJobTimeOut(currentTime, last_alarm_time, Integer.parseInt(alarm_limit_time.trim()));
                    if (geFlag) {
                        LOGGER.info("spark队列在queue_memory_alarm中有该类型告警记录,本次告警距离上次告警超过了指定间隔时间,将告警:" + alarm_content);
                        try {
                            MySQLUtils.executeUpdate(connection, addSql);
                        }catch (Exception e){
                            LOGGER.error("spark队列告警失败!!!");
                        }
                    }else {//本次告警距离上次告警没超过指定间隔时间，但保存记录alarm_type类型改为2
                        MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
                    }
                }
                else {//该类型的队列还没有告警过,直接告警,并保存告警记录
                    try {
                        LOGGER.info("spark队列该类型首次告警成功...");
                        MySQLUtils.executeUpdate(connection, addSql);
                    }catch (Exception e){
                        LOGGER.error("spark队列告警发送短信失败!!!");
                    }
                }
                MySQLUtils.returnConnection(connection);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(4,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",2)";
            Connection connection = null;
            try {
                connection = MySQLUtils.getConnection();
                MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
            }catch (Exception e){
                e.printStackTrace();
            }
            LOGGER.info("spark队列本次统计比例没有超过指定的比例...");
        }
    }


    /**
     * dev 队列类型统计告警
     * @param currentTime
     * @param ge_used_min_memory
     */

    private static void devAlarm(String currentTime,String ge_used_min_memory,String batch_number) {

        int memory_limit = Integer.parseInt(PropertiesUtil.get("dev_memory_limit").trim());
        int total_memory_limit_dev = Integer.parseInt(PropertiesUtil.get("total_memory_limit_dev"));
        String[] memory = ge_used_min_memory.split("\t");
        String used_mem = memory[0];
        String min_mem = memory[1];
        if(Integer.parseInt(used_mem.trim())<memory_limit && Integer.parseInt(used_mem.trim())<Integer.parseInt(min_mem.trim())){
            LOGGER.info("dev队列使用数没有超过配置中数,current_used_memory:"+used_mem+" config_memory_limit:"+memory_limit+"  min_mem:"+min_mem);
            return;
        }

        float devRate = Float.parseFloat(used_mem)/Float.parseFloat(min_mem);
        String twoPoint_rate = UUIDByDateUtils.twoPointNumbers(devRate+"");
        String rate = twoPoint_rate.replaceAll("\\.", "");
        int cal_rate = Integer.parseInt(rate);
        int self_rate = Integer.parseInt(PropertiesUtil.get("dev_rate").trim());
        String alarm_limit_time = PropertiesUtil.get("queue_alarm_limit_time").trim();
        if(cal_rate>self_rate) {
            //告警的手机号
            String alarm_list = PropertiesUtil.get("alarm_phone_list").trim();
            String alarm_title = "dev队列使用memory过高";
            String alarm_content = "dev队列使用memory过高,使用:" + used_mem + " ,总数为:" + min_mem + " ,所占比例为:" + rate + "%";

            //判断在告警记录表(job_source_alarm)中是否有同类型的ge告警过,如果alarm_time到当前时间超过了配置中限制时间alarm_limit_time则告警,
            // 否则不同时满足比例大于指定比例,间隔时间不大于指定时间则不告警
            //表jobs_source_alarm告警类型;1表示ge,2表示alliance,3表示mem,4表示spark,5表示dev,6表示recommed,7表示analysis
            String sql = "SELECT j.id,j.alarm_time FROM queue_memory_alarm j WHERE j.is_alarm=1 and j.queue_id=5 ORDER BY id DESC LIMIT 1";
            Connection connection = null;
            ResultSet rs = null;
            try {
                connection = MySQLUtils.getConnection();
                rs = MySQLUtils.query(connection, sql);
                rs.next();
                int rows = rs.getRow();
                //大于0说明jobs_source_alarm中已经告警过该类型的memory,则根据alarm_time与当前时间判断是否超过了配置中设置的间隔时间alarm_limit_time,超过则再一次告警
                //超过后告警成功则把新的告警记录保存下来,没超过就不用管了
                String addSql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(5,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",1)";
                String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(5,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",2)";

                if (rows > 0) {
                    String last_alarm_time = rs.getString("alarm_time");
                    boolean geFlag = jdugeJobTimeOut(currentTime, last_alarm_time, Integer.parseInt(alarm_limit_time.trim()));
                    if (geFlag) {
                        LOGGER.info("dev队列在queue_memory_alarm中有该类型告警记录,本次告警距离上次告警超过了指定间隔时间,将告警:" + alarm_content);
                        try {
                            MySQLUtils.executeUpdate(connection, addSql);
                        }catch (Exception e){
                            LOGGER.error("dev队列告警失败!!!");
                        }
                    }else {//本次告警距离上次告警没超过指定间隔时间，但保存记录alarm_type类型改为2
                        MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
                    }
                }
                else {//该类型的队列还没有告警过,直接告警,并保存告警记录
                    try {
                        LOGGER.info("dev队列该类型首次告警成功...");
                        MySQLUtils.executeUpdate(connection, addSql);
                    }catch (Exception e){
                        LOGGER.error("dev队列告警发送短信失败!!!");
                    }
                }
                MySQLUtils.returnConnection(connection);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(5,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",2)";
            Connection connection = null;
            try {
                connection = MySQLUtils.getConnection();
                MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
            }catch (Exception e){
                e.printStackTrace();
            }
            LOGGER.info("dev队列本次统计比例没有超过指定的比例...");
        }
    }


    /**
     * recommed 队列类型统计告警
     * @param currentTime
     * @param ge_used_min_memory
     */

    private static void recommedAlarm(String currentTime,String ge_used_min_memory,String batch_number) {

        int memory_limit = Integer.parseInt(PropertiesUtil.get("recommed_memory_limit").trim());
        int total_memory_limit_recommed = Integer.parseInt(PropertiesUtil.get("total_memory_limit_recommed"));
        String[] memory = ge_used_min_memory.split("\t");
        String used_mem = memory[0];
        String min_mem = memory[1];
        if(Integer.parseInt(used_mem.trim())<memory_limit && Integer.parseInt(used_mem.trim())<Integer.parseInt(min_mem.trim())){
            LOGGER.info("recommed队列使用数没有超过配置中数,current_used_memory:"+used_mem+" config_memory_limit:"+memory_limit+"  min_mem:"+min_mem);
            return;
        }

        float recommedRate = Float.parseFloat(used_mem)/Float.parseFloat(min_mem);
        String twoPoint_rate = UUIDByDateUtils.twoPointNumbers(recommedRate+"");
        String rate = twoPoint_rate.replaceAll("\\.", "");
        int cal_rate = Integer.parseInt(rate);
        int self_rate = Integer.parseInt(PropertiesUtil.get("recommed_rate").trim());
        String alarm_limit_time = PropertiesUtil.get("queue_alarm_limit_time").trim();
        if(cal_rate>self_rate) {
            //告警的手机号
            String alarm_list = PropertiesUtil.get("alarm_phone_list").trim();
            String alarm_title = "recommed队列使用memory过高";
            String alarm_content = "recommed队列使用memory过高,使用:" + used_mem + " ,总数为:" + min_mem + " ,所占比例为:" + rate + "%";

            //判断在告警记录表(job_source_alarm)中是否有同类型的ge告警过,如果alarm_time到当前时间超过了配置中限制时间alarm_limit_time则告警,
            // 否则不同时满足比例大于指定比例,间隔时间不大于指定时间则不告警
            //表jobs_source_alarm告警类型;1表示ge,2表示alliance,3表示mem,4表示spark,5表示dev,6表示recommed,7表示analysis
            String sql = "SELECT j.id,j.alarm_time FROM queue_memory_alarm j WHERE j.is_alarm=1 and j.queue_id=6 ORDER BY id DESC LIMIT 1";
            Connection connection = null;
            ResultSet rs = null;
            try {
                connection = MySQLUtils.getConnection();
                rs = MySQLUtils.query(connection, sql);
                rs.next();
                int rows = rs.getRow();
                //大于0说明jobs_source_alarm中已经告警过该类型的memory,则根据alarm_time与当前时间判断是否超过了配置中设置的间隔时间alarm_limit_time,超过则再一次告警
                //超过后告警成功则把新的告警记录保存下来,没超过就不用管了
                String addSql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(6,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",1)";
                String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(6,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",2)";

                if (rows > 0) {
                    String last_alarm_time = rs.getString("alarm_time");
                    boolean geFlag = jdugeJobTimeOut(currentTime, last_alarm_time, Integer.parseInt(alarm_limit_time.trim()));
                    if (geFlag) {
                        LOGGER.info("recommed队列在queue_memory_alarm中有该类型告警记录,本次告警距离上次告警超过了指定间隔时间,将告警:" + alarm_content);
                        try {
                            MySQLUtils.executeUpdate(connection, addSql);
                        }catch (Exception e){
                            LOGGER.error("recommed队列告警失败!!!");
                        }
                    }else {//本次告警距离上次告警没超过指定间隔时间，但保存记录alarm_type类型改为2
                        MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
                    }
                }
                else {//该类型的队列还没有告警过,直接告警,并保存告警记录
                    try {
                        LOGGER.info("recommed队列该类型首次告警成功...");
                        MySQLUtils.executeUpdate(connection, addSql);
                    }catch (Exception e){
                        LOGGER.error("recommed队列告警发送短信失败!!!");
                    }
                }
                MySQLUtils.returnConnection(connection);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(6,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",2)";
            Connection connection = null;
            try {
                connection = MySQLUtils.getConnection();
                MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
            }catch (Exception e){
                e.printStackTrace();
            }
            LOGGER.info("recommed队列本次统计比例没有超过指定的比例...");
        }
    }


    /**
     * analysis 队列类型统计告警
     * @param currentTime
     * @param ge_used_min_memory
     */

    private static void analysisAlarm(String currentTime,String ge_used_min_memory,String batch_number) {

        int memory_limit = Integer.parseInt(PropertiesUtil.get("analysis_memory_limit").trim());
        int total_memory_limit_analysis = Integer.parseInt(PropertiesUtil.get("total_memory_limit_analysis"));
        String[] memory = ge_used_min_memory.split("\t");
        String used_mem = memory[0];
        String min_mem = memory[1];
        if(Integer.parseInt(used_mem.trim())<memory_limit && Integer.parseInt(used_mem.trim())<Integer.parseInt(min_mem.trim())){
            LOGGER.info("analysis队列使用数没有超过配置中数,current_used_memory:"+used_mem+" config_memory_limit:"+memory_limit+"  min_mem:"+min_mem);
            return;
        }


        float analysisRate = Float.parseFloat(used_mem)/Float.parseFloat(min_mem);
        String twoPoint_rate = UUIDByDateUtils.twoPointNumbers(analysisRate+"");
        String rate = twoPoint_rate.replaceAll("\\.", "");
        int cal_rate = Integer.parseInt(rate);
        int self_rate = Integer.parseInt(PropertiesUtil.get("recommed_rate").trim());
        String alarm_limit_time = PropertiesUtil.get("queue_alarm_limit_time").trim();
        if(cal_rate>self_rate) {
            //告警的手机号
            String alarm_list = PropertiesUtil.get("alarm_phone_list").trim();
            String alarm_title = "analysis队列使用memory过高";
            String alarm_content = "analysis队列使用memory过高,使用:" + used_mem + ",总数为:" + min_mem + " ,所占比例为:" + rate + "%";

            //判断在告警记录表(job_source_alarm)中是否有同类型的ge告警过,如果alarm_time到当前时间超过了配置中限制时间alarm_limit_time则告警,
            // 否则不同时满足比例大于指定比例,间隔时间不大于指定时间则不告警
            //表jobs_source_alarm告警类型;1表示ge,2表示alliance,3表示mem,4表示spark,5表示dev,6表示recommed,7表示analysis
            String sql = "SELECT j.id,j.alarm_time FROM queue_memory_alarm j WHERE j.is_alarm=1 and j.queue_id=7 ORDER BY id DESC LIMIT 1";
            Connection connection = null;
            ResultSet rs = null;
            try {
                connection = MySQLUtils.getConnection();
                rs = MySQLUtils.query(connection, sql);
                rs.next();
                int rows = rs.getRow();
                //大于0说明jobs_source_alarm中已经告警过该类型的memory,则根据alarm_time与当前时间判断是否超过了配置中设置的间隔时间alarm_limit_time,超过则再一次告警
                //超过后告警成功则把新的告警记录保存下来,没超过就不用管了
                String addSql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(7,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",1)";
                String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(7,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",2)";

                if (rows > 0) {
                    String last_alarm_time = rs.getString("alarm_time");
                    boolean geFlag = jdugeJobTimeOut(currentTime, last_alarm_time, Integer.parseInt(alarm_limit_time.trim()));
                    if (geFlag) {
                        LOGGER.info("analysis队列在queue_memory_alarm中有该类型告警记录,本次告警距离上次告警超过了指定间隔时间,将告警:" + alarm_content);
                        try {
                            MySQLUtils.executeUpdate(connection, addSql);
                        }catch (Exception e){
                            LOGGER.error("analysis队列告警失败!!!");
                        }
                    }else {//本次告警距离上次告警没超过指定间隔时间，但保存记录alarm_type类型改为2
                        MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
                    }
                }
                else {//该类型的队列还没有告警过,直接告警,并保存告警记录
                    try {
                        LOGGER.info("analysis队列该类型首次告警成功...");
                        MySQLUtils.executeUpdate(connection, addSql);
                    }catch (Exception e){
                        LOGGER.error("analysis队列告警发送短信失败!!!");
                    }
                }
                MySQLUtils.returnConnection(connection);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            String add_limit_alarm_Sql = "INSERT INTO queue_memory_alarm (queue_id,alarm_time,alarm_limit_time,used_memory,total_memory,cal_rate,self_rate,memory_limit,batch_number,is_alarm) VALUES(7,'" + currentTime + "','" + alarm_limit_time + "'," + used_mem + "," + min_mem + "," + cal_rate + ","+ self_rate + ","+ memory_limit + ","+ batch_number + ",2)";
            Connection connection = null;
            try {
                connection = MySQLUtils.getConnection();
                MySQLUtils.executeUpdate(connection, add_limit_alarm_Sql);
            }catch (Exception e){
                e.printStackTrace();
            }
            LOGGER.info("analysis队列本次统计比例没有超过指定的比例...");
        }
    }

    public static String countQueuesNum(String sql,Connection connection) throws  Exception{
        ResultSet rs = MySQLUtils.query(connection,sql);
        rs.next();
        int rows = rs.getRow();
        String sumnums = "0";
        if(rows>0){
            sumnums = rs.getString(1);
        }else{
            LOGGER.info("sql没有查到数据:"+sql);
        }
        return  sumnums;
    }

    public static String get_used_min_memory(String sql,Connection connection) throws  Exception{
        ResultSet rs = MySQLUtils.query(connection,sql);
        rs.next();
        int rows = rs.getRow();
        String used_min_memory = "0";
        if(rows>0){
            String used_memory = rs.getString(1);
            String min_memory = rs.getString(2);
            used_min_memory = used_memory +"\t"+ min_memory;
        }else{
            LOGGER.info("sql没有查到数据:"+sql);
        }
        return  used_min_memory;
    }





    /**
     * 判断时间是否超时
     * @param current_time
     * @param last_alarm_time
     * @param limit_time
     * @return
     */
    public static  boolean  jdugeJobTimeOut(String current_time,String last_alarm_time,int limit_time){

        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
        boolean flag = false;//默认开始没有超时
        try {
            Date new_date = sdf.parse(current_time.trim());
            long new_time = new_date.getTime();

            Date old_date = sdf.parse(last_alarm_time.trim());
            long old_time = old_date.getTime();

            long exe_time = (new_time-old_time)/1000;
            if(exe_time>limit_time){//超时
                flag = true;
            }
        } catch (ParseException e) {
            e.printStackTrace();
            throw new RuntimeException("判断队列memory比例时间是否超过上次告警时间...异常:"+e.getMessage());
        }
        return flag;
    }

}
